/*
 * Copyright (c) 2019.  drakelee.base All rights reserved.
 */

package com.base.components.task.service;

import com.base.components.common.boot.EventHandler;
import com.base.components.common.boot.SpringContextUtil;
import com.base.components.common.exception.business.BusinessException;
import com.base.components.common.service.task.SysTask;
import com.base.components.common.util.JsonUtils;
import com.base.components.task.exception.DestroyingException;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Uninterruptibles;
import org.quartz.CronScheduleBuilder;
import org.quartz.CronTrigger;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 作业管理
 *
 * @author <a href="drakelee1221@gmail.com">LiGeng</a>
 * @version 1.0.0, 2017-08-03 15:04
 *
 */
@Service
@RefreshScope
public class TaskManager {

  private static Logger logger = LoggerFactory.getLogger(TaskManager.class);

  /**
   * 当前服务器任务集合
   */
  private static final Map<String, SysTask> DOING_TASK = Maps.newConcurrentMap();

  /**
   * 任务组名
   */
  public static final String GROUP_NAME = "DYNAMIC_TASK";

  /**
   * 被手动触发的任务参数KEY
   */
  public static final String TRIGGER_TASK_KEY = "TRIGGER_TASK_KEY";

  /**
   * 是否已初始化
   */
  private static AtomicReference<Boolean> initialization = new AtomicReference<>(Boolean.FALSE);

  /**
   * 是否在销毁中
   */
  private static AtomicReference<Boolean> destroying = new AtomicReference<>(Boolean.FALSE);

  /**
   * 当触发的作业被其它服务器执行时的等待间隔时间
   */
  @Value("${base.quartz.lockingSleepIntervalMillisecond:2000}")
  private long lockingSleepIntervalMillisecond;

  /**
   * 当触发的作业被其它服务器执行时的最大等待时间
   */
  @Value("${base.quartz.lockingSleepMillisecond:600000}")
  private long lockingMaxMillisecond;

  @Value("${server.port}")
  private String serverPort;
  /**
   * 当前server: localhost:port
   */
  public static String serverHost;

  @Autowired
  private SysTaskService taskService;

  @Autowired
  private Scheduler scheduler;

  @PostConstruct
  private void registerInit(){
    SpringContextUtil.addStartedEvents(new EventHandler<ApplicationContext, Object>() {
      @Override
      public String getId() {
        return "TaskManager registerInit";
      }
      @Override
      public Object onEvent(ApplicationContext applicationContext) {
        init();
        return null;
      }
    });
  }

  private void init() {
    if(!initialization.get()){
      logger.info("--------正在初始化动态调度任务--------");
      try {
        serverHost = InetAddress.getLocalHost().getHostAddress()+":"+serverPort;
      } catch (UnknownHostException ignore) {
        serverHost = "http://localhost:"+serverPort;
      }
      List<SysTask> list = findEnableTasks();
      int count = reBuildCurrentServer(list);
      try {
        if (!scheduler.isStarted()) {
          scheduler.start();
        }
      } catch (Exception e) {
        logger.error("start scheduler error", e);
      }
      logger.info("--------动态调度任务初始化成功({})--------", count);
      logger.info("--------{}--------", serverHost);
      initialization.set(Boolean.TRUE);
    }
  }

  public List<SysTask> findEnableTasks() {
    return taskService.findEnableTasks();
  }

  public synchronized int reBuildCurrentServer(List<SysTask> list) {
    int count = 0;
    for (SysTask task : list) {
      try {
        boolean result = addJob(task);
        if (result) {
          logger.info("构建动态调度任务: {}, {}", ++count ,task);
        }
      } catch (Throwable e) {
        e.printStackTrace();
      }
    }
    return count;
  }

  /**
   * 在 TaskDestroyProcessKillForcedShutdownEvent 中调用，
   * 为保证在关闭事件中优先执行-不要实现 DisposableBean
   **/
  public synchronized void destroy() {
    if(scheduler == null){
      return;
    }
    try {
      logger.info("----------动态调度任务暂停并销毁 >>> 开始--------------");
      destroying.set(Boolean.TRUE);
      while (!DOING_TASK.isEmpty()){
        Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
      }
      try {
        scheduler.pauseAll();
      } catch (SchedulerException ignore) {
      }
      scheduler.shutdown(true);
    } catch (Exception ignore) {
    }finally {
      scheduler = null;
      destroying.set(Boolean.FALSE);
      initialization.set(Boolean.FALSE);
      logger.info("----------动态调度任务暂停并销毁 >>> 结束--------------");
    }
  }

  public void removeJob(String taskId, boolean waiting) {
    try {
      if(waiting){
        waitingBlockingDoingTask(taskId);
      }
      TriggerKey triggerKey = TriggerKey.triggerKey(taskId, GROUP_NAME);
      JobKey jobKey = JobKey.jobKey(taskId, GROUP_NAME);
      scheduler.pauseTrigger(triggerKey);
      scheduler.unscheduleJob(triggerKey);
      scheduler.deleteJob(jobKey);
    } catch (Exception e) {
      throw new BusinessException(e, 0);
    }
  }

  public void removeJob(String taskId) {
    removeJob(taskId, true);
  }

  public void triggerTask(String taskId, Map<String, Object> argumentsMap) {
    try {
      waitingBlockingDoingTask(taskId);
      JobKey jobKey = JobKey.jobKey(taskId, GROUP_NAME);
      Assert.isTrue(scheduler.checkExists(jobKey),
                    "当前服务中未注册此作业[taskId="+taskId+"]，请先将此作业同步到当前服务中");
      JobDataMap map = new JobDataMap(argumentsMap);
      map.put(TRIGGER_TASK_KEY, Boolean.TRUE);
      scheduler.triggerJob(jobKey, map);
    } catch (Exception e) {
      throw new BusinessException(e, 0);
    }
  }

  @SuppressWarnings("unchecked")
  private static JobDetail buildJob(SysTask task) {
    try {
      Class<? extends Job> cl = (Class<? extends Job>) Class.forName(task.getTargetClass());
      DynamicJobDetail jd = new DynamicJobDetail();
      jd.setJobClass(cl);
      jd.setName(task.getId());
      jd.setGroup(GROUP_NAME);
      if (!Strings.isNullOrEmpty(task.getArgumentsMap())) {
        Map<String, Object> map = Maps.newHashMap();
        try {
          map.putAll(JsonUtils.reader(task.getArgumentsMap(), HashMap.class));
        } catch (IOException e) {
          e.printStackTrace();
        }
        jd.setJobDataMap(new JobDataMap(map));
      }
      return jd;
    } catch (Exception e) {
      logger.error("--- buildJob ---", e);
    }
    return null;
  }

  private CronTrigger buildTrigger(SysTask task) {
    CronScheduleBuilder scheduleBuilder =
        CronScheduleBuilder.cronSchedule(task.getCronExpression());
    //超过触发时间不做任何事情
    scheduleBuilder.withMisfireHandlingInstructionDoNothing();
    CronTrigger tri = TriggerBuilder.newTrigger().withIdentity(task.getId(), GROUP_NAME)
                                    .withSchedule(scheduleBuilder).build();
    Date next = tri.getFireTimeAfter(null);
    if (next == null || next.getTime() <= System.currentTimeMillis()) {
      task.setDisabled(true);
      taskService.saveOrUpdate(task);
      return null;
    }
    return tri;
  }



//  /**
//   * 停止或启动某一服务器动态作业功能
//   *
//   * @param server
//   * @param stop
//   * @param async
//   */
//  public static void stopTaskWithBroadcast(ServerRegister server, boolean stop, boolean async) {
//    try {
//      ServerRegisterService srs =
//          ApplicationContext.getCurrent().getBean(ServerRegisterService.class);
//      TaskBroadcastServlet.BROADCAST_TYPE type = null;
//      if (stop) {
//        type = TaskBroadcastServlet.BROADCAST_TYPE.STOP_ALL;
//      } else {
//        type = TaskBroadcastServlet.BROADCAST_TYPE.START_ALL;
//      }
//      if (async) {
//        srs.sendAsyncBroadcast(server, null, type);
//      } else {
//        srs.sendBroadcast(server, null, type);
//      }
//    } catch (Throwable e) {
//      e.printStackTrace();
//    }
//  }

  public static void sendBroadcast(String fireTaskId, TaskBroadcastType type,
                                   boolean async) {
//    ServerRegisterService srs =
//        ApplicationContext.getCurrent().getBean(ServerRegisterService.class);
//    if (async) {
//      srs.sendAsyncBroadcast(fireTaskId, type);
//    } else {
//      srs.sendBroadcast(fireTaskId, type);
//    }
  }


  /**
   * @param task
   * @return true=添加成功，false=添加失败
   * @throws Throwable
   */
  public boolean addJob(SysTask task) throws Throwable {
    if(task.getDisabled()){
      return false;
    }
    waitingBlockingDoingTask(task.getId());
    CronTrigger tri = buildTrigger(task);
    JobDetail job = buildJob(task);
    if (tri == null || job == null) {
      return false;
    }
    Trigger storeTri = scheduler.getTrigger(tri.getKey());
    if (storeTri != null) {
      removeJob(storeTri.getKey().getName());
    }
    scheduler.scheduleJob(job, tri);
    return true;
  }


//  private static ServerRegister shutdownServer(String serverHost, String serverName) {
//    try {
//      ServerRegisterService srService =
//          ApplicationContext.getCurrent().getBean(ServerRegisterService.class);
//      ServerRegister sr = new ServerRegister();
//      sr.setStopTask(TASK_STOP);
//      sr.setName(serverName);
//      sr.setHost(serverHost);
//      sr.setShutdown(true);
//      return srService.saveServer(sr);
//    } catch (Throwable e) {
//      e.printStackTrace();
//    }
//    return null;
//  }

//  private static ServerRegister registServer(String serverHost, String serverName) {
//    try {
//      ServerRegisterService srService =
//          ApplicationContext.getCurrent().getBean(ServerRegisterService.class);
//      ServerRegister sr = new ServerRegister();
//      sr.setStopTask(TASK_STOP);
//      sr.setName(serverName);
//      sr.setHost(serverHost);
//      return srService.saveServer(sr);
//    } catch (Throwable e) {
//      e.printStackTrace();
//    }
//    return null;
//  }

//  public static void setStopTask(boolean stop) {
//    TASK_STOP = false;
//    ServerRegisterService srService =
//        ApplicationContext.getCurrent().getBean(ServerRegisterService.class);
//    if (stop) {
//      destroy(false);
//      srService.updateStopTask(CURRENT_SERVERHOST, true);
//    } else {
//      reBuildCurrentServer(findUndisabledTasks());
//      srService.updateStopTask(CURRENT_SERVERHOST, false);
//    }
//    TASK_STOP = stop;
//    stopLog();
//  }
//
//  private static void stopLog() {
//    if (TASK_STOP) {
//      logger.warn("【动态作业已停止工作！】");
//    } else {
//      logger.warn("【动态作业已开始工作！】");
//    }
//  }


  /**
   * 阻塞等待当前正在调用的作业对象，等待执行完毕
   * @param taskId
   */
  public void waitingBlockingDoingTask(String taskId){
    SysTask task = DOING_TASK.get(taskId);
    long interval = 500;
    long waitMillis = 0;
    while(null != task && waitMillis < lockingMaxMillisecond){
      Uninterruptibles.sleepUninterruptibly(interval, TimeUnit.MILLISECONDS);
      waitMillis += interval;
      task = DOING_TASK.get(taskId);
    }
  }

  public static void addDoingTask(SysTask doingTask){
    if(destroying.get()){
      throw new DestroyingException("任务调度器正在停止中，不接受执行新的任务！");
    }
    DOING_TASK.put(doingTask.getId(), doingTask);
  }

  public static void removeDoingTask(String doingTaskId){
    DOING_TASK.remove(doingTaskId);
  }

  public long getLockingSleepIntervalMillisecond() {
    return lockingSleepIntervalMillisecond;
  }

  public long getLockingMaxMillisecond() {
    return lockingMaxMillisecond;
  }
}
